package club.lovety.code.rocketmq_demo;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageQueueListener;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;
import java.util.Set;

/**
 * <p>
 * 这里写有象视频代码描述
 * </p>
 *
 * @author sunmc
 * @date created in 2018/5/3 10:10
 * @since
 */
public class ConsumerDemo {


    public static void main(String[] args) throws Exception {
//        pullConsumer();
        pushConsumer();
    }


    public  static void pushConsumer() throws  Exception{

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("last_offset_new_consumer");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe("mmm", "*");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                MessageExt messageExt = msgs.get(0);
                System.out.println(Thread.currentThread().getName()+"content: "+ new String(messageExt.getBody())+",");
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.out.printf("Consumer Started.%n");
    }



    public  static void pullConsumer() throws Exception {

        DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer("smc_pull_consumer");

        defaultMQPullConsumer.setNamesrvAddr("localhost:9876");

        defaultMQPullConsumer.registerMessageQueueListener("smc",new MessageQueueListener(){

            @Override
            public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
                System.out.println("topic: "+ topic+" , message: ");
            }
        });

        defaultMQPullConsumer.start();


    }


}
